package com.hex.ds.hdtp.core.inf.data.contrast.service.impl;

import com.hex.bigdata.hbda.utils.StringUtils;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcConnectUtil;
import com.hex.ds.hdtp.core.inf.metadata.entity.TableFieldsEntity;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.ISourceConfInfoRepository;
import com.hex.ds.hdtp.core.inf.metadata.service.impl.ODPSMethodService;
import com.hex.ds.hdtp.core.inf.metadata.service.impl.SourceMethodHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Service("COMPARE_ODPS")
public class OdpsInfoService extends AbstractDataInfoService {
    private final ODPSMethodService odpsMethodService;

    public static final String COLUMN_NAME = "column_name";

    public OdpsInfoService(ISourceConfInfoRepository sourceConfInfoRepository, SourceMethodHandlerContext sourceMethodHandlerContext, ODPSMethodService odpsMethodService) {
        super(sourceConfInfoRepository, sourceMethodHandlerContext);
        this.odpsMethodService = odpsMethodService;
    }

    @Override
    public List<String> selectPartitionField(String sourceCode, String schema, String table) throws SQLException {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        String querySql = odpsMethodService.getTablePartitionFieldSql(schema, table);
        List<String> allFieldsName = JdbcConnectUtil.getFieldResultList(sourceConfInfoPoList, querySql);
        return allFieldsName.subList(0, allFieldsName.indexOf("#ROWS"));
    }

    /**
     * @param sourceCode
     * @param schema
     * @param table
     * @return
     * @throws SQLException
     * @Description 所有字段类型
     */
    @Override
    public List<Map<String, String>> selectAllFieldType(String sourceCode, String schema, String table) throws SQLException {
        List<TableFieldsEntity> tableFieldsEntities = odpsMethodService.tableStructure(sourceCode, schema, table);
        List<Map<String, String>> result = tableFieldsEntities.stream().map(tableFieldsEntity -> {
            Map<String, String> fieldMap = new HashMap<>();
            fieldMap.put(COLUMN_NAME, tableFieldsEntity.getColumnName().toUpperCase());
            fieldMap.put("data_type", tableFieldsEntity.getColumnType().toUpperCase());
            return fieldMap;
        }).collect(Collectors.toList());
        return result;
    }

    @Override
    public List<Map<String, String>> selectAllFieldLength(String sourceCode, String schema, String table) throws SQLException {
        List<TableFieldsEntity> tableFieldsEntities = odpsMethodService.tableStructure(sourceCode, schema, table);
        //所有字段的长度
        List<Map<String, String>> result = tableFieldsEntities.stream().map(tableFieldsEntity -> {
            Map<String, String> fieldMap = new HashMap<>();
            fieldMap.put(COLUMN_NAME, tableFieldsEntity.getColumnName().toUpperCase());
            fieldMap.put("data_length", StringUtils.isBlank(tableFieldsEntity.getColumnLength()) ? "0" : tableFieldsEntity.getColumnLength());
            return fieldMap;
        }).collect(Collectors.toList());
        return result;
    }

    @Override
    public List<Map<String, String>> selectFieldAccuracy(String sourceCode, String schema, String table) throws SQLException {
        List<TableFieldsEntity> tableFieldsEntities = odpsMethodService.tableStructure(sourceCode, schema, table);
        //所有字段的精度
        List<Map<String, String>> result = tableFieldsEntities.stream().map(tableFieldsEntity -> {
            Map<String, String> fieldMap = new HashMap<>();
            fieldMap.put(COLUMN_NAME, tableFieldsEntity.getColumnName().toUpperCase());
            if (StringUtils.isBlank(tableFieldsEntity.getColumnLength()) || "0.".equals(tableFieldsEntity.getColumnLength())) {
                fieldMap.put("column_accuracy", StringUtils.isBlank(tableFieldsEntity.getColumnLength()) ? "0" : tableFieldsEntity.getColumnLength());
            } else {
                String[] split = tableFieldsEntity.getColumnLength().split(",");
                fieldMap.put("column_accuracy", split.length == 1 ? split[0] : split[1]);
            }
            return fieldMap;
        }).collect(Collectors.toList());
        return result;
    }


    @Override
    protected String selectDataNumSql(String schema, String table, String whereSql) {
        return odpsMethodService.getTableDataNumSql(schema, table, whereSql);
    }


    @Override
    protected String selectCountSumIsNullSql(String schema, String table, String field, String whereSql) {
        if (whereSql != null && whereSql.trim().toUpperCase().startsWith("WHERE")) {
            whereSql = "AND " + whereSql.trim().substring(5);
        }
        return odpsMethodService.getTableIsNullCountSql(schema, table, field, whereSql);
    }
}

